package org.gxlu.com.jstorm.day1;

import java.util.Map;

import backtype.storm.task.OutputCollector;
import backtype.storm.task.TopologyContext;
import backtype.storm.topology.OutputFieldsDeclarer;
import backtype.storm.topology.base.BaseRichBolt;
import backtype.storm.tuple.Fields;
import backtype.storm.tuple.Tuple;
import backtype.storm.tuple.Values;

public class SplitBolt extends BaseRichBolt {

	/**
	 * 
	 */
	private static final long serialVersionUID = 1L;
	private OutputCollector collector = null ;
	@Override
	public void execute(Tuple tuple) {
		 String str = tuple.getString(0);
         String[] array = str.split(" ") ;
         for(String s: array){
        	 if(!s.equals(""))
        	   collector.emit(new Values(s,1)) ;
         }
		// collector.emit(new Values(str));
	}

	@Override
	public void prepare(Map arg0, TopologyContext arg1, OutputCollector arg2) {
		 System.out.println("---StringWopriBolt--prepare-") ;
		 collector = arg2 ;

	}

	@Override
	public void declareOutputFields(OutputFieldsDeclarer declare) {
		 declare.declare(new Fields("word","num"));

	}

}